#include "RelOp.h"

void Sum::Run(Pipe &inPipe, Pipe &outPipe, Function &computeMe){

    in = &inPipe;
    out = &outPipe;
    func = &computeMe;

    pthread_create(&thread, NULL, SumWorker, (void*)this);
}

void Sum::WaitUntilDone(){
    pthread_join(thread, NULL);
}

void Sum::Use_n_Pages(int n){
    nOfPages = n;
}

void Sum::DoWork(){

    Record temp;
    int intResult;
    double doubleResult;
    double result = 0;
    Type t;

    while(in->Remove(&temp)){
        t = func->Apply(temp, intResult, doubleResult);

        if(t == Int)
            result += intResult;
        else
            result += doubleResult;
    }

    Attribute a = {"double", Double};
    Schema schema ("schema", 1, &a);

    std::stringstream s;
    s << result << "|";

    std::string str = s.str();
    char * writable = new char[str.size() + 1];
    std::copy(str.begin(), str.end(), writable);
    writable[str.size()] = '\0';

    Record r;
    r.ComposeRecord(&schema, writable);

    out->Insert(&r);
    out->ShutDown();

    delete[] writable;
}

void *SumWorker(void *param){
     Sum sum = *((Sum*) param);
     sum.DoWork();
}
